package com.azier.influxdb.util;

import com.azier.influxdb.util.InfluxDBHandleUtil;
import com.azier.util.DateUtil;
import okhttp3.OkHttpClient;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBException;
import org.influxdb.InfluxDBFactory;
import org.influxdb.InfluxDBMapperException;
import org.influxdb.annotation.Column;
import org.influxdb.dto.*;
import org.influxdb.impl.InfluxDBMapper;

import java.lang.reflect.Field;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.select;

/**
 * InfluxDB数据库连接操作类
 *
 * @author 罗军
 * res_URL:https://blog.csdn.net/x541211190/article/details/83216589
 */
public class InfluxDBConnectionUtil {

    // 用户名
    private String username;
    // 密码
    private String password;
    // 连接地址
    private String openurl;

    // 数据库
    private String database;
    // 保留策略
    private String retentionPolicy;

    private InfluxDB influxDB;

    private boolean isEnableBatch;

    private int maxPonits;

    private int maxTime;

    private long connectTimeOut;

    private long readTimeOut;

    private long writeTimeOut;

    private InfluxDBMapper influxDBMapper;
    public InfluxDBConnectionUtil(String username, String password, String openurl, String database,
                                  String retentionPolicy) {
        this.username = username;
        this.password = password;
        this.openurl = openurl;
        this.database = database;
        this.retentionPolicy = retentionPolicy == null || retentionPolicy.equals("") ? "autogen" : retentionPolicy;
        influxDbBuild();
        influxDbMapperBuild();
    }
    public InfluxDBConnectionUtil(String username, String password, String openurl, String database,
                                  String retentionPolicy,boolean isEnableBatch,int maxPonits,int maxTime,
                                  long connectTimeOut,long readTimeOut,long writeTimeOut) {
        this.username = username;
        this.password = password;
        this.openurl = openurl;
        this.database = database;
        this.connectTimeOut = connectTimeOut;
        this.readTimeOut = readTimeOut;
        this.writeTimeOut = writeTimeOut;
        this.retentionPolicy = retentionPolicy == null || retentionPolicy.equals("") ? "autogen" : retentionPolicy;
        influxDbBuild();
        influxDbMapperBuild();
        setEnableBatch(isEnableBatch,maxPonits,maxTime);
    }
    public String getDatabase() {
        return database;
    }
    public InfluxDB getInfluxDB() {
        return influxDB;
    }

    public InfluxDBMapper getInfluxDBMapper() {
        return influxDBMapper;
    }
    /**
     * 创建数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public void createDB(String dbName) {
        influxDB.createDatabase(dbName);
    }

    /**
     * 删除数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public void deleteDB(String dbName) {
        influxDB.deleteDatabase(dbName);
    }

    /**
     * 测试连接是否正常
     *
     * @return true 正常
     */
    public boolean ping() {
        boolean isConnected = false;
        Pong pong;
        try {
            pong = influxDB.ping();
            if (pong != null) {
                isConnected = true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return isConnected;
    }

    /**
     * 连接时序数据库
     *
     * @return
     */
    private InfluxDB influxDbBuild() {
        if (influxDB == null) {
//            influxDB = InfluxDBFactory.connect(openurl, username, password);
            OkHttpClient.Builder okHttpBuilder =new OkHttpClient.Builder()
                .connectTimeout(connectTimeOut, TimeUnit.SECONDS)
                .readTimeout(readTimeOut, TimeUnit.SECONDS)
                .writeTimeout(writeTimeOut, TimeUnit.SECONDS);
            influxDB = InfluxDBFactory.connect(openurl, username, password,okHttpBuilder);
            influxDB.setLogLevel(InfluxDB.LogLevel.NONE);
        }
        return influxDB;
    }

    /**
     * 设定启用
     * @param isEnableBatch
     * @param maxPonits
     * @param maxTime
     */
    public void setEnableBatch(boolean isEnableBatch,int maxPonits,int maxTime){
        this.isEnableBatch = isEnableBatch;
        this.maxPonits = maxPonits;
        this.maxTime = maxTime;
        setEnableBatch();
    }
    public void setEnableBatch(){
        if(influxDB.isBatchEnabled()){
            influxDB.disableBatch();
        }
        if(isEnableBatch){
            influxDB.enableBatch(maxPonits,maxTime,TimeUnit.MILLISECONDS);
        }
    }
    /**
     * 创建influxDBMapper
     *
     * @return
     */
    private InfluxDBMapper influxDbMapperBuild() {
        if (influxDBMapper == null) {
            influxDBMapper = new InfluxDBMapper(influxDB);
        }
        return influxDBMapper;
    }
    /**
     * 创建自定义保留策略
     *
     * @param policyName
     *            策略名
     * @param duration
     *            保存天数
     * @param replication
     *            保存副本数量
     * @param isDefault
     *            是否设为默认保留策略
     */
    public void createRetentionPolicy(String policyName, String duration, int replication, Boolean isDefault) {
        String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s ", policyName,
            database, duration, replication);
        if (isDefault) {
            sql = sql + " DEFAULT";
        }
        this.query(sql);
    }

    /**
     * 创建默认的保留策略
     *
     * 策略名：default，保存天数：30天，保存副本数量：1
     *            设为默认保留策略
     */
    public void createDefaultRetentionPolicy() {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
            "default", database, "30d", 1);
        this.query(command);
    }

    /**
     * 查询
     *
     * @param command
     *            查询语句
     * @return
     */
    public QueryResult query(String command)throws InfluxDBException {
        return influxDB.query(new Query(command, database));
    }

    /**
     * 插入
     *
     * @param measurement
     *            表
     * @param tags
     *            标签
     * @param fields
     *            字段
     */
    public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields, long time,
                       TimeUnit timeUnit) throws InfluxDBException{
        Point.Builder builder = Point.measurement(measurement);
        builder.tag(tags);
        builder.fields(fields);
        if (0 != time) {
            builder.time(time, timeUnit);
        }
        influxDB.write(database, retentionPolicy, builder.build());
    }

    /**
     * 批量写入测点
     *
     * @param batchPoints
     */
    public void batchInsert(BatchPoints batchPoints)throws InfluxDBException {
        influxDB.write(batchPoints);
        // influxDB.enableGzip();
        // influxDB.enableBatch(2000,100,TimeUnit.MILLISECONDS);
        // influxDB.disableGzip();
        // influxDB.disableBatch();
    }

    /**
     * 批量写入数据
     *
     * @param database
     *            数据库
     * @param retentionPolicy
     *            保存策略
     * @param consistency
     *            一致性
     * @param records
     *            要保存的数据（调用BatchPoints.lineProtocol()可得到一条record）
     */
    public void batchInsert(final String database, final String retentionPolicy, final InfluxDB.ConsistencyLevel consistency,
                            final List<String> records) throws InfluxDBException{
        influxDB.write(database, retentionPolicy, consistency, records);
    }

    /**
     * 删除
     *
     * @param command
     *            删除语句
     * @return 返回错误信息
     */
    public String deleteMeasurementData(String command) throws InfluxDBException{
        QueryResult result = influxDB.query(new Query(command, database));
        return result.getError();
    }

    /**
     * 关闭数据库
     */
    public void close() {
        influxDB.close();
    }

    /**
     * 构建Point
     *
     * @param measurement
     * @param time
     * @param fields
     * @return
     */
    public Point pointBuilder(String measurement, long time, Map<String, String> tags, Map<String, Object> fields)throws InfluxDBException {
        Point point = Point.measurement(measurement).time(time, TimeUnit.MILLISECONDS).tag(tags).fields(fields).build();
        return point;
    }

    /**
     * 执行指定的查询，并返回指定类型的结果集
     * @param query
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> List<T> query(Query query, Class<T> clazz) throws InfluxDBException{
        return influxDBMapper.query(query,clazz);
    }

    /**
     * 根据所给出的类（非对象），组装查询语句（全表查询），并返回指定类型的结果集
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> List<T> query(Class<T> clazz) throws InfluxDBException{
        return influxDBMapper.query(clazz);
    }
    /**
     * 存入一个记录
     * @param <T> model
     * @return
     */
    public <T> void save(T model) throws InfluxDBException{
        influxDBMapper.save(model);
    }

    public <T> List<T> selectAll(Class<T> clazz) throws InfluxDBException{
        String table = InfluxDBHandleUtil.getMeasurementName(clazz);
        Query select = select().all().from(database, table);
        return influxDBMapper.query(select,clazz);
    }

//    public <T> List<T> selectByObject() throws InfluxDBException{
//        String table = InfluxDBHandleUtil.getMeasurementName(clazz);
//        Query select = select().all().from(database, table);
//        return influxDBMapper.query(select,clazz);
//    }

    /**
     * 根据给定的实体查询，所有条件使用等于判断
     * @param model
     * @param <T>
     * @return
     * @throws InfluxDBException
     */
    public <T> List<T> selectByObject(T model) throws InfluxDBException{
        try {
            StringBuilder select = new StringBuilder("SELECT * FROM ");
            //--以下通过反射从注解获取字段名，字段类型，库名和表名
            Class cls = model.getClass();
            String table = InfluxDBHandleUtil.getMeasurementName(cls);
            select.append(table).append(" ");
            String database = InfluxDBHandleUtil.getDatabaseName(cls);
            //反射获取类的Column的注解信息
            ConcurrentMap<String, Field> colNameAndFieldMap = InfluxDBHandleUtil.getColumn(cls);
            int i = 0;
            for (String key : colNameAndFieldMap.keySet()) {

                Field field = colNameAndFieldMap.get(key);
                Column column = field.getAnnotation(Column.class);
                //字段名
                String columnName = column.name();
                //是否是tag
                boolean isTag = column.tag();
                Class<?> fieldType = field.getType();

                if (!field.isAccessible()) {
                    field.setAccessible(true);
                }
                //字段值
                Object value = field.get(model);

                if (value != null) {
                    if ("time".equals(columnName)) {
                        //time暂时不作为条件
                        if (value != null) {
                            if(i>0){
                                select.append("AND ");
                            }else if(i==0){
                                select.append("WHERE ");
                            }
                            InfluxDBHandleUtil.setTime(select,fieldType,value);
                        }
                    } else{
                        if(i>0){
                            select.append("AND ");
                        }else if(i==0){
                            select.append("WHERE ");
                        }
                        InfluxDBHandleUtil.setField(select,fieldType,columnName,value);
                        //标记位自增
                        i++;
                    }
                }

            }
            System.out.println(select.toString());
            Query query = new Query(select.toString(),database);
            return query(query,cls);
        } catch (IllegalAccessException e) {
            throw new InfluxDBMapperException(e);
        }
    }
    public QueryResult test(){
        Date date = DateUtil.offsetDay(DateUtil.parseDateTime("2019-06-05 16:54:00"),3);
        Instant instant = Instant.ofEpochMilli(date.getTime());
        DateTimeFormatter formatter = DateTimeFormatter
            .ofPattern("yyyy-MM-dd'T'HH:mm:ss'Z'")
            .withZone(ZoneId.of("UTC"));
        String timeP1 = formatter.format(Instant.ofEpochMilli(date.getTime()));
        Query query = new Query("SELECT * FROM demo where time='"+timeP1+"'",database);
        QueryResult results = influxDB.query(query);
        return results;
    }
}


